package doct.mobi.album.myrxjava;

import android.content.Context;
import android.util.Log;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.io.BufferedReader;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.schedulers.Schedulers;

/**
 * Created by huangli on 17/12/9.
 */

public class RxManager {

    public static final String TAG = "RxManager";

    /**
     * 一个响应式拉去实例
     */
    public static void flowableResponsiveDemo(final Context context){
        Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> emitter) throws Exception {
                Log.i(TAG,"First requested = " + emitter.requested());
                try{
                    InputStream inputStream = context.getResources().getAssets().open("test.txt");
                    InputStreamReader inputStreamReader = new InputStreamReader(inputStream);
                    BufferedReader br = new BufferedReader(inputStreamReader);
                    String str;
                    while ((str = br.readLine()) != null && !emitter.isCancelled()) {
                        while (emitter.requested() == 0) {
                            if (emitter.isCancelled()) {
                                break;
                            }
                        }
                        emitter.onNext(str);
                        Log.d(TAG, "emit " + str + " , requested = " + emitter.requested());
                    }
                    br.close();
                    inputStreamReader.close();
                    emitter.onComplete();
                }catch (Exception e){
                    emitter.onError(e);
                }
            }
        },BackpressureStrategy.ERROR).subscribeOn(Schedulers.io()).observeOn(Schedulers.newThread()).subscribe(new Subscriber<String>() {

            private Subscription mSubscription;

            @Override
            public void onSubscribe(Subscription s) {
                mSubscription = s;
                s.request(1);
            }

            @Override
            public void onNext(String string) {
                System.out.println(string);
                try {
                    Thread.sleep(2000);
                    mSubscription.request(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    /**
     * FlowableEmitter中requested接口完成响应式拉取
     */
    public static void flowableEmitterResponsive(Subscriber<Integer> downstream){
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG,"First requested = " + emitter.requested());
                boolean flag;
                for (int i = 0; ; i++){
                    flag = false;
                    while (emitter.requested() == 0) {
                        if (!flag) {
                            Log.d(TAG, "Oh no! I can't emit value!");
                            flag = true;
                        }
                    }
                    emitter.onNext(i);
                    Log.d(TAG, "emit " + i + " , requested = " + emitter.requested());
                }
            }
        },BackpressureStrategy.ERROR).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(downstream);
    }

    /**
     * FlowableEmitter中requested接口作用 异步的情况
     */
    public static void flowableEmitterReqAsync(){
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                Log.d(TAG, "current requested: " + emitter.requested());
            }
        },BackpressureStrategy.ERROR).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                Log.d(TAG, "onSubscribe");
                s.request(1000); //我要打1000个！！
            }

            @Override
            public void onNext(Integer integer) {
                Log.d(TAG, "onNext: " + integer);
            }

            @Override
            public void onError(Throwable t) {
                Log.w(TAG, "onError: ", t);
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "onComplete");
            }
        });
    }

    /**
     * FlowableEmitter中requested接口作用 同步的情况
     */
    public static void flowableEmitterReqSync(){
        Flowable.create(new FlowableOnSubscribe<Integer>() {

            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
//                Log.i(TAG, "current requested: " + emitter.requested());

                Log.i(TAG, "before emit, requested = " + emitter.requested());

                Log.i(TAG, "emit 1");
                emitter.onNext(1);
                Log.i(TAG, "after emit 1, requested = " + emitter.requested());

                Log.i(TAG, "emit 2");
                emitter.onNext(2);
                Log.i(TAG, "after emit 2, requested = " + emitter.requested());

                Log.i(TAG, "emit 3");
                emitter.onNext(3);
                Log.i(TAG, "after emit 3, requested = " + emitter.requested());

                /** ----- 这里如果发送大于Subscriber中的request数量，将会发生错误 ----- **/
                for (int i = 0; i < 10; i++){
                    Log.i(TAG, "emit "+i);
                    emitter.onNext(i);
                    Log.i(TAG, "after emit complete, requested = " + emitter.requested());
                }
                /** ----- 这里如果发送大于Subscriber中的request数量，将会发生错误 ----- **/

                Log.i(TAG, "emit complete");
                emitter.onComplete();

                Log.i(TAG, "after emit complete, requested = " + emitter.requested());
            }
        },BackpressureStrategy.ERROR).subscribe(new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                s.request(1); //我要打十个！
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG,"onNext : "+integer);
            }

            @Override
            public void onError(Throwable t) {
                Log.i(TAG,"onError : "+t.getLocalizedMessage());
            }

            @Override
            public void onComplete() {
                Log.i(TAG,"onComplete");
            }
        });
    }

    /**
     * Flowable的Backpressure策略 之 Latest
     */
    public static void flowbaleLatest(Subscriber<Integer> downstream){
        Flowable<Integer> upstream = Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                for (int i = 0;i < 9999; i++) {     //注意这里是128，超过128会报错
                    Log.d(TAG, "emit " + i);
                    emitter.onNext(i);
                }
            }
        }, BackpressureStrategy.LATEST);
        upstream.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(downstream);
    }

    /**
     * Flowable的Backpressure策略 之 DROP
     */
    public static void flowbaleDrop(Subscriber<Integer> downstream){
        Flowable<Integer> upstream = Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                for (int i = 0;i < 9999; i++) {     //注意这里是128，超过128会报错
                    Log.d(TAG, "emit " + i);
                    emitter.onNext(i);
                }
            }
        }, BackpressureStrategy.DROP);
        upstream.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(downstream);
    }


    /**
     * 大水缸的oom
     */
    public static void flowableBuffer(){
        Flowable<Integer> upstream = Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                for (int i = 0;; i++) {     //注意这里是128，超过128会报错
                    Log.d(TAG, "emit " + i);
                    emitter.onNext(i);
                }
            }
        }, BackpressureStrategy.BUFFER);
        Subscriber<Integer> downstream = new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                Log.i(TAG,"onSubscribe");
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG,"onNext : "+integer);
            }

            @Override
            public void onError(Throwable t) {
                Log.i(TAG,"onError : "+t.getLocalizedMessage());
            }

            @Override
            public void onComplete() {
                Log.i(TAG,"onComplete");
            }
        };
        upstream.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(downstream);
    }

    /**
     * Flowable 128缓存
     */
    public static void flowable128(){
        Flowable<Integer> upstream = Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                for (int i = 0; i < 128; i++) {     //注意这里是128，超过128会报错
                    Log.d(TAG, "emit " + i);
                    emitter.onNext(i);
                }
            }
        }, BackpressureStrategy.ERROR);  //上下游流速不均匀的时候抛出 MissingBackpressureException
        Subscriber<Integer> downstream = new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                Log.i(TAG,"onSubscribe");
//                s.request(Long.MAX_VALUE);      //这里注释掉响应式拉取也是没问题的，应该不同线程Flowable会有一个水缸缓存
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG,"onNext : "+integer);
            }

            @Override
            public void onError(Throwable t) {
                Log.i(TAG,"onError : "+t.getLocalizedMessage());
            }

            @Override
            public void onComplete() {
                Log.i(TAG,"onComplete");
            }
        };
        upstream.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(downstream);
    }

    /**
     * Flowable在不同线程中的表现
     */
    public static void flowableThread(){
        Flowable<Integer> upstream = Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                Log.d(TAG, "emit 1");
                emitter.onNext(1);
                Log.d(TAG, "emit 2");
                emitter.onNext(2);
                Log.d(TAG, "emit 3");
                emitter.onNext(3);
                Log.d(TAG, "emit complete");
                emitter.onComplete();
            }
        }, BackpressureStrategy.ERROR);  //上下游流速不均匀的时候抛出 MissingBackpressureException
        Subscriber<Integer> downstream = new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                Log.i(TAG,"onSubscribe");
//                s.request(Long.MAX_VALUE);      //这里注释掉响应式拉取也是没问题的，应该不同线程Flowable会有一个水缸缓存
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG,"onNext : "+integer);
            }

            @Override
            public void onError(Throwable t) {
                Log.i(TAG,"onError : "+t.getLocalizedMessage());
            }

            @Override
            public void onComplete() {
                Log.i(TAG,"onComplete");
            }
        };
        upstream.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(downstream);
    }

    /**
     * Flowable
     */
    public static void flowableBase(){
        Flowable<Integer> upstream = Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                Log.d(TAG, "emit 1");
                emitter.onNext(1);
                Log.d(TAG, "emit 2");
                emitter.onNext(2);
                Log.d(TAG, "emit 3");
                emitter.onNext(3);
                Log.d(TAG, "emit complete");
                emitter.onComplete();
            }
        }, BackpressureStrategy.ERROR);  //上下游流速不均匀的时候抛出 MissingBackpressureException

        Subscriber<Integer> downstream = new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                Log.i(TAG,"onSubscribe");
//                s.request(Long.MAX_VALUE);      //响应式拉取,如果注释掉这行会抛出 MissingBackpressureException
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG,"onNext : "+integer);
            }

            @Override
            public void onError(Throwable t) {
                Log.i(TAG,"onError : "+t.getLocalizedMessage());
            }

            @Override
            public void onComplete() {
                Log.i(TAG,"onComplete");
            }
        };
        upstream.subscribe(downstream);
    }

    /**
     * 从速度上进行治理, 减缓事件发送进水缸的速度
     */
    public static void relieveBackPressureDelay(){
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                for (int i = 0; ; i++) {
                    emitter.onNext(i);
                    Thread.sleep(2000);  //每次发送完事件延时2秒
                }
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "" + integer);
                    }
         });
    }

    /**
     * 从定时采样上进行治理, 减少发送进水缸里的事件
     */
    public static void relieveBackPressureSample(){
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                for (int i = 0; ; i++) {
                    emitter.onNext(i);
                }
            }
        }).subscribeOn(Schedulers.io()).sample(2, TimeUnit.SECONDS)  //sample取样
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "" + integer);
                    }
        });
    }

    /**
     * 从数量上进行治理, 减少发送进水缸里的事件
     */
    public static void relieveBackPressureFilter(){
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                for (int i = 0; ; i++) {
                    emitter.onNext(i);
                }
            }
        }).subscribeOn(Schedulers.io()).filter(new Predicate<Integer>() {
            @Override
            public boolean test(Integer integer) throws Exception {
                return integer % 10 == 0;
            }
        }).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d(TAG, "" + integer);
            }
        });
    }

    public static void backPressureDemo2(){
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                for (int i = 0; ; i++) {   //无限循环发事件
                    emitter.onNext(i);
                    Log.d(TAG,"发送 "+i);
                }
            }
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer o) throws Exception {
                Thread.sleep(2000);
                Log.d(TAG, "＊＊＊＊＊＊＊＊＊＊＊＊收到＊＊＊＊＊＊＊＊＊＊＊＊＊" + o);
            }
        });
    }

    public static void backPressureDemo1(){
        Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                for (int i = 0; ; i++){
                    emitter.onNext(i);
                    Log.d(TAG, "发送 "+i);
                }
            }
        }).subscribeOn(Schedulers.io());

        Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                emitter.onNext("A");
            }
        }).subscribeOn(Schedulers.io());

        Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {
            @Override
            public String apply(Integer integer, String s) throws Exception {
                return integer + s;
            }
        }).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<String>() {
            @Override
            public void accept(String o) throws Exception {
                Log.d(TAG, o);
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                Log.w(TAG, throwable);
            }
        });
    }

    public static void rxZip(){
        Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.d(TAG, "emit 1");
                emitter.onNext(1);
                Thread.sleep(1000);
                Log.d(TAG, "emit 2");
                emitter.onNext(2);
                Thread.sleep(1000);
                Log.d(TAG, "emit 3");
                emitter.onNext(3);
                Thread.sleep(1000);
                Log.d(TAG, "emit 4");
//                emitter.onNext(4);
//                Thread.sleep(1000);
//                Log.d(TAG, "emit complete1");
                emitter.onComplete();
            }
        }).subscribeOn(Schedulers.newThread());
        Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                Log.d(TAG, "emit A");
                emitter.onNext("A");
                Thread.sleep(1000);
                Log.d(TAG, "emit B");
                emitter.onNext("B");
                Thread.sleep(1000);
                Log.d(TAG, "emit C");
                emitter.onNext("C");
                Thread.sleep(1000);
                Log.d(TAG, "emit complete2");
                emitter.onComplete();
            }
        }).subscribeOn(Schedulers.newThread());

        Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {
            @Override
            public String apply(Integer integer, String s) throws Exception {
                return integer + s;
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "onSubscribe");
            }

            @Override
            public void onNext(String value) {
                Log.d(TAG, "onNext: " + value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "onError");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "onComplete");
            }
        });
    }

    public static void rxConcatMap(){
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).concatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                final List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("I am value " + integer);
                }
                return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, s);
            }
        });
    }

    public static void rxFlatMap(){
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).flatMap(new Function<Integer, ObservableSource<String>>() {

            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                final List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("I am value " + integer);
                }
                return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, s);
            }
        });

    }

    public static void rxJavaMap(){
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                return "result : "+integer;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.i(TAG,s);
            }
        });
    }

    public static void rxJavaThread_Demo2(){
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onComplete();
                Log.i(TAG, "Observable thread is : " + Thread.currentThread().getName());
            }
        });
        Consumer<Integer> consumer = new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i(TAG, "Observer thread is :" + Thread.currentThread().getName());
                Log.i(TAG, "onNext: " + integer);
            }
        };
        observable.subscribeOn(Schedulers.newThread()).subscribeOn(AndroidSchedulers.mainThread()).observeOn(AndroidSchedulers.mainThread()).
                doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.i(TAG, "After observeOn(mainThread), current thread is: " + Thread.currentThread().getName());
                    }
                }).observeOn(Schedulers.newThread()).doOnNext(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i(TAG, "After observeOn(newThread), current thread is: " + Thread.currentThread().getName());
            }
        }).subscribe(consumer);
    }

    /**
     * subscribeOn()只有第一次有效
     * observeOn()
     */
    public static void rxJavaThread_Demo1(){
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onComplete();
                Log.i(TAG, "Observable thread is : " + Thread.currentThread().getName());
            }
        });
        Consumer<Integer> consumer = new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i(TAG, "Observer thread is :" + Thread.currentThread().getName());
                Log.i(TAG, "onNext: " + integer);
            }
        };
        observable.subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread()).subscribe(consumer);
    }

    public static void rxJavaBase() {
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onComplete();
            }
        });

        Observer<Integer> observer = new Observer<Integer>() {

            private Disposable disposable;

            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "onSubscribe");
                disposable = d;
            }

            @Override
            public void onNext(Integer value) {
                Log.i(TAG, "onNext " + value);
                if (value == 2){
                    disposable.dispose();
                }
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "onError");
            }

            @Override
            public void onComplete() {
                Log.i(TAG, "onComplete");
            }
        };
        observable.subscribe(observer);
    }
}
